package com.lagou.mall.esflow.server.util;

import static java.util.Optional.ofNullable;

import com.alibaba.fastjson.JSONObject;
import com.lagou.mall.esflow.client.Binlog;
import com.lagou.mall.esflow.server.ESFlowSystemException;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.StringUtils;

public class KafkaMsgUtil {

    public static Binlog parseBinlogData(String json) {
        try {
//            String json = new String(msg, StandardCharsets.UTF_8);
            Binlog binlog = JsonUtil.parseFromJsonStr(json, Binlog.class);

            JSONObject oldData = binlog.getOld();
            JSONObject newData = binlog.getData();

            binlog.setOld(phraseBinlogDataWithDump(oldData));
            binlog.setData(phraseBinlogDataWithDump(newData));

            return binlog;
        } catch (Exception e) {
            throw ESFlowSystemException.error("parseBinlog failed", e);
        }
    }


    /**
     * 将binlog中记录的整行的数据转化为对应的驼峰的格式
     *
     * @param data binlog中对应的行操作记录
     */
    public static JSONObject phraseBinlogDataWithDump(JSONObject data) {
        JSONObject dataWithDump = new JSONObject();
        ofNullable(data).ifPresent(jsonObject -> jsonObject.entrySet().forEach(
                stringObjectEntry -> {
                    String newKey = fieldToDump(stringObjectEntry.getKey());
                    dataWithDump.put(newKey, stringObjectEntry.getValue());
                }
        ));
        return dataWithDump;
    }

    /**
     * 将数据库中的字段改变为驼峰命名格式
     *
     * @param field 数据库的字段名
     */
    private static String fieldToDump(String field) {
        return Arrays.stream(field.split("_")).reduce(null, (left, right) -> {
            if (null == left) {
                return right;
            }
            if (StringUtils.isBlank(right)) {
                return left;
            }
            String first = String.valueOf(right.charAt(0));
            return left + right.replaceFirst(first, first.toUpperCase());
        });
    }


    public static Object getKey(String field, Binlog binlog) {
        JSONObject data = binlog.getData();
        AtomicReference<String> result = new AtomicReference<>();
        ofNullable(field)
                .ifPresent(fieldName -> result.set(String.valueOf(data.get(field))));
        return result.get();
    }


}
